package com.xxl.job.executor.util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.executor.annotation.LogTime;
import com.xxl.job.executor.dto.ApiInfoDto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.net.URI;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author dingfeng
 * @version 2.0
 * @dec 任务处理工具类
 * @date 2022/3/3 15:03
 */
@Component
@Slf4j
public class TaskUtilV2 {
    @Value("${api.info.mode:public}")
    private String apInfoMode;

    @Value("${task.main.thread:5}")
    private Integer taskMainThread;

    @Value("${task.son.thread:20}")
    private Integer taskSonThread;

    @Value("${pageSize:2000}")
    private Integer pageSize;

    @Value("${appCode}")
    private String appCode;

    @Resource
    private JdbcTemplate jdbcTemplate;

    /** 主任务处理线程池 */
    private ExecutorService mainExecutorService;

    @PostConstruct
    public void init() {
        // 初始化主任务线程池
        mainExecutorService = Executors.newFixedThreadPool(taskMainThread);
    }

    /**
     * 主任务处理方法
     * @param params 调度中心传过来的参数 包含模式名 + 表名
     */
    public void handler(String params) throws Exception {
        String[] tableNames = params.split(",");
        // 主任务结果集合
        List<Future<TaskReturn>> allTaskReturn = new ArrayList();
        // 失败主任务结果集合
        List<TaskReturn> allFailTaskReturn = new ArrayList();
        for (int i = 0; i < tableNames.length; i++) {
            log.info("开始执行表名为==》{}的调度",tableNames[i]);
            Future<TaskReturn> submit = mainExecutorService.submit(new Task(tableNames[i].trim()));
            allTaskReturn.add(submit);
        }

//        for (String tableName : tableNames) {
//            Future<TaskReturn> submit = mainExecutorService.submit(new Task(tableName.trim()));
//            allTaskReturn.add(submit);
//        }
        while (allTaskReturn.size() != 0) {
            Future<TaskReturn> taskReturnFuture = allTaskReturn.remove(0);
            TaskReturn taskReturn = null;
            taskReturn = taskReturnFuture.get();
            if (taskReturn.isSuccess) {
                log.info("表{}插入成功，总计数据{}条，耗时{}秒。", taskReturn.getTableName(), taskReturn.getRows(), taskReturn.getTime());
            } else {
                log.error("表{}插入失败，失败原因", taskReturn.getTableName(), taskReturn.getException().getCause());
                allFailTaskReturn.add(taskReturn);
            }
        }

        if (allFailTaskReturn.size() > 0) {
            StringBuilder failTaskTableNames = new StringBuilder("执行失败的表:\n");
            while (allFailTaskReturn.size() != 0) {
                TaskReturn remove = allFailTaskReturn.remove(0);
                failTaskTableNames.append(remove.getTableName() + "\n");
            }
            /**
             * 方案一： 让调度中心获取异常信息
             * 好处：无需自己编码处理重试
             * 缺点.调度中心的失败重试机制 无法修改参数
             *   导致失败重试会重新执行所有表 操作耗时
             * 方案二：自己编码处理异常
             * 好处：可以单独处理失败的表，粒度小，执行快
             * 缺点：自己编码实现
             * 方案三： 在调度中心配置的时候每一张表都单独配置一个任务
             * 好处：失败重试也只会重试这一张表,
             * 缺点：配置繁琐
             */
            //方案1
            XxlJobHelper.handleFail(failTaskTableNames.toString());
            //方案2
//          ArrayList<String> retryMessageArray = new ArrayList<>();
//          retry(allFailTaskReturn, retryMessageArray, 2,true);
//          XxlJobHelper.handleFail(retryMessageArray.toString());
//          方案3
        }
    }

    /**
     * 失败任务重试 递归处理
     */
    public List retry(List<TaskReturn> allFailTaskReturn, List retryMessageArray, Integer retryNumber, Boolean firstTime) {
        if (retryNumber == 0 || allFailTaskReturn.size() == 0) {
            return retryMessageArray;
        }
        if (firstTime) {
            for (TaskReturn taskReturn : allFailTaskReturn) {
                retryMessageArray.add(taskReturn.getTableName() + "执行失败");
            }
        }
        ArrayList<Future<TaskReturn>> allTaskReturn = new ArrayList<>();
        ArrayList<TaskReturn> tempFailTaskReturn = new ArrayList<>();
        for (int i = 0; i < allFailTaskReturn.size(); i++) {
            TaskReturn taskReturn = allFailTaskReturn.get(i);
            String tableName = taskReturn.getTableName();
            Future<TaskReturn> submit = mainExecutorService.submit(new Task(tableName));
            allTaskReturn.add(submit);
        }

        for (int i = 0; i < allTaskReturn.size(); i++) {
            Future<TaskReturn> taskReturnFuture = allTaskReturn.get(i);
            try {
                TaskReturn taskReturn = taskReturnFuture.get();
                if (!taskReturn.isSuccess) {
                    tempFailTaskReturn.add(taskReturn);
                    retryMessageArray.add(allFailTaskReturn.get(i).getTableName() + "重试失败");
                    continue;
                }
                retryMessageArray.add(allFailTaskReturn.get(i) + "重试成功");
            } catch (Exception e) {
                TaskReturn taskReturn = new TaskReturn().setIsSuccess(false)
                        .setTableName(allFailTaskReturn.get(i).getTableName())
                        .setException(e);
                tempFailTaskReturn.add(taskReturn);
            }
        }
        return retry(tempFailTaskReturn, retryMessageArray, --retryNumber, false);
    }


    /**
     * 分发批量插入请求
     *
     * @param apiInfoDto
     * @throws Exception
     */
    @LogTime
    public TaskReturn distributeHttpGet(ApiInfoDto apiInfoDto) throws Exception {
        /**
         * 先发送一次请求
         * 1.可以判断接口是否通
         * 2.获取到表的数据总量 分批插入
         */
        HashMap<String, String> requestParams = new HashMap<>(32);
        requestParams.put("appCode", appCode);
        requestParams.put("pageNum", "1");
        requestParams.put("pageSize", "1");
        JSONObject parseObject;
        JSONObject o;
        String res;
        try {
            res = httpGet(apiInfoDto.getApiPath(), requestParams);
            parseObject = JSONObject.parseObject(res);
            Map<String, Object> innerMap = parseObject.getInnerMap();
            o = (JSONObject) innerMap.get("data");
        } catch (Exception e) {
            return new TaskReturn()
                    .setException(e)
                    .setIsSuccess(false)
                    .setTableName(apiInfoDto.getTableName());
        }
        // 数据总数
        Integer totalNum = (Integer) o.get("totalNum");
        // 阿里中台有限制，一次最多只返回2000条数据
        Integer distributeNumber = totalNum % pageSize == 0 ? totalNum / pageSize : totalNum / pageSize + 1;
        // 回滚标志 AtomicBoolean保证线程安全
        AtomicBoolean rollbackFlag = new AtomicBoolean(false);
        // 记录第一个失败的线程的失败原因
        List<Throwable> firstFail = new ArrayList();
        // 主线程 栅栏
        CountDownLatch mainThreadLatch = new CountDownLatch(1);
        /**
         * 子任务执行成功数量 AtomicInteger保证线程安全
         * 当 atomicInteger 等于 distributeNumber 代表请求都已经成功
         * 通知主线程 栅栏放行
         */
        AtomicInteger atomicInteger = new AtomicInteger(0);
        ExecutorService distributeExecutorService = Executors.newFixedThreadPool(taskSonThread);
        // 任务开始时间
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < distributeNumber; i++) {
            if (rollbackFlag.get()) {
                return new TaskReturn().setIsSuccess(false);
            }
            ApiInfoDto tempApiInfoDto = new ApiInfoDto();
            BeanUtils.copyProperties(apiInfoDto, tempApiInfoDto);
            tempApiInfoDto.setPageNum(i + 1);
            distributeExecutorService.submit(new DistributeTask(mainThreadLatch, rollbackFlag, tempApiInfoDto, distributeNumber, atomicInteger, firstFail));
        }
        /**
         * 等待放行
         *  1. 任务全部执行成功 子任务通知放行
         *  2. 有一个子任务执行失败，子任务马上通知放行
         */
        mainThreadLatch.await();
        // 关闭线程池，释放资源
        distributeExecutorService.shutdown();
        // 任务结束时间
        long endTime = System.currentTimeMillis();
        TaskReturn taskReturn = new TaskReturn()
                .setTableName(apiInfoDto.getTableName())
                .setIsSuccess(!rollbackFlag.get())
                .setRows(totalNum).setTime((endTime - startTime) / 1000);
        if (!taskReturn.isSuccess) {
            if (firstFail.size() != 0) {
                taskReturn.setException(firstFail.get(0));
            }
        }
        return taskReturn;
    }

    /**
     * 处理htppGet请求
     */
    @LogTime
    public  String httpGet(String apiPath, Map<String, String> params) throws Exception {
        CloseableHttpClient httpClient = HttpPool.getHttpClient();
        URIBuilder uriBuilder = new URIBuilder(apiPath);
        // 定义请求的参数
        if (null != params) {
            Iterator<String> iterator = params.keySet().iterator();
            while (iterator.hasNext()) {
                String key = iterator.next();
                uriBuilder.setParameter(key, params.get(key));
            }
        }
        URI uri = uriBuilder.build();
        HttpGet request = new HttpGet(uri);
        request.setHeader("accept", "*/*");
        CloseableHttpResponse response = null;
        try {
            // 执行请求
            response = httpClient.execute(request);
            // 判断返回状态是否为200
            if (response.getStatusLine().getStatusCode() == 200) {
                //请求体内容
                String content = EntityUtils.toString(response.getEntity(), "UTF-8");
                //内容写入文件
                return content;
            } else {
                log.error("请求{}失败，状态码:{}", request.getRequestLine(), response.getStatusLine().getStatusCode()
                        );
                return "";
            }
        } catch (Exception ex) {
            throw ex;
        } finally {
            if (response != null) {
                response.close();
            }
            //相当于关闭浏览器
            //httpClient.close();
        }
    }

    /**
     * 解析res,生成sql
     *
     * @param res       response返回结果 json字符串
     * @param tableName 表名
     * @return sql
     */
    @LogTime
    public  String handlerResToGetSql(String res, String tableName) {
        //判断fieldNames是否已经拼接好了
        Boolean fieldNamesState = false;
        String resSql = "insert into " + tableName;
        StringBuilder fieldNames = new StringBuilder("(");
        StringBuilder fieldValues = new StringBuilder();
        JSONObject parseObject = JSONObject.parseObject(res);
        Map<String, Object> innerMap = parseObject.getInnerMap();
        JSONObject o = (JSONObject) innerMap.get("data");
        JSONArray rows = (JSONArray) o.get("rows");
        for (int i = 0; i < rows.size(); i++) {
            StringBuffer tempValue = new StringBuffer("(");
            JSONObject obj = rows.getJSONObject(i);
            Set<String> keys = obj.keySet();
            for (String fieldName : keys) {
                if (!fieldNamesState) {
                    fieldNames.append("\"" + fieldName + "\",");
                }
                Object fieldValue = obj.get(fieldName);
                if (fieldValue == null || fieldValue instanceof Number) {
                    if (fieldValue instanceof BigDecimal) {
                        fieldValue = ((BigDecimal) fieldValue).toPlainString();
                    }
                    tempValue.append(fieldValue + ",");
                } else {
                    tempValue.append("'" + fieldValue + "',");
                }
            }
            if (!fieldNamesState) {
                fieldNamesState = true;
            }
            fieldValues.append(tempValue.substring(0, tempValue.length() - 1) + "),");
        }
        String fieldNamesRes = fieldNames.substring(0, fieldNames.length() - 1) + ")";
        String fieldValuesRes = fieldValues.substring(0, fieldValues.length() - 1) + ";";

        return resSql + fieldNamesRes + " values" + fieldValuesRes;
    }

    /**
     * 主任务类
     */
    class Task implements Callable<TaskReturn> {
        private String tableName;

        public Task(String tableName) {
            this.tableName = tableName;
        }

        @Override
        public TaskReturn call() throws Exception {
            // 1.解析表名称
            if (tableName.indexOf(".") == -1) {
                log.error("表名{}错误,请检查是否有带上模式名", tableName);
                return new TaskReturn()
                        .setIsSuccess(false)
                        .setException(new RuntimeException("表名" + tableName + "错误,请检查是否有带上模式名"))
                        .setTableName(tableName);
            }
            TaskReturn taskReturn = null;
            try {
                // 2.获取表的api相关信息
                String getApiInfoSql = "select * from " + apInfoMode + ".api_info where table_name = ?";
                String[] args = new String[]{tableName};
                Map<String, Object> apiInfoMap = jdbcTemplate.queryForMap(getApiInfoSql, args);
                // 3. 执行前
                Object beforeSql = apiInfoMap.get("before_sql");
                if (!Objects.isNull(beforeSql) && ((String) beforeSql).length() > 0) {
                    jdbcTemplate.update((String) beforeSql);
                    log.info("执行前语句:" + beforeSql + "执行成功");
                }
                // 4. 执行
                ApiInfoDto apiInfoDto = new ApiInfoDto()
                        .setTableName((String) apiInfoMap.get("table_name"))
                        .setApiPath((String) apiInfoMap.get("api_path"))
                        .setPageSize(pageSize);
                taskReturn = distributeHttpGet(apiInfoDto);
                if (!taskReturn.isSuccess) {
                    log.error("子任务执行出错,清空表数据");
                    jdbcTemplate.update("TRUNCATE TABLE " + apiInfoDto.getTableName());
                    // 清空表后,终止程序运行
                    return taskReturn;
                }
                //执行后语句
                Object afterSql = apiInfoMap.get("after_sql");
                if (!Objects.isNull(afterSql) && ((String) afterSql).length() > 0) {
                    jdbcTemplate.update((String) afterSql);
                    log.info("执行后语句:" + afterSql + "执行成功");
                }
            } catch (Exception e) {
                return new TaskReturn()
                        .setIsSuccess(false)
                        .setException(e)
                        .setTableName(tableName);
            }
            return taskReturn;
        }
    }

    /**
     * 子任务类
     */
    class DistributeTask implements Callable<Object> {
        // 回滚标志
        private final AtomicBoolean rollbackFlag;
        private final ApiInfoDto apiInfoDto;
        // 主线程栅栏
        private final CountDownLatch mainThreadLatch;
        private final Integer distributeNumber;
        private final AtomicInteger successNumber;
        private final List<Throwable> firstFail;

        public DistributeTask(CountDownLatch mainThreadLatch, AtomicBoolean rollbackFlag
                , ApiInfoDto apiInfoDto, Integer distributeNumber
                , AtomicInteger successNumber, List<Throwable> firstFail) {
            this.mainThreadLatch = mainThreadLatch;
            this.rollbackFlag = rollbackFlag;
            this.apiInfoDto = apiInfoDto;
            this.distributeNumber = distributeNumber;
            this.successNumber = successNumber;
            this.firstFail = firstFail;

        }

        @Override
        public Object call() throws Exception {
            //封装请求参数
            HashMap<String, String> requestParams = new HashMap<>(8);
            requestParams.put("appCode", appCode);
            requestParams.put("pageNum", apiInfoDto.getPageNum() + "");
            requestParams.put("pageSize", apiInfoDto.getPageSize() + "");
            /**
             * 判断回滚标志
             * 1.如果为true,则不再请求
             *  但是在高并发情况下 不能控制代码的原子性
             *  需要在finally里再次判断,如果回滚标志被设置为true
             *  则需要执行相应的操作删除脏数据
             */
            if (rollbackFlag.get()) {
                return false;
            }
            //发送请求
            String res = httpGet(apiInfoDto.getApiPath(), requestParams);
            //解析出sql
            String sql = handlerResToGetSql(res, apiInfoDto.getTableName());
            //执行sql
            try {
                Long start = System.currentTimeMillis();
                int update = jdbcTemplate.update(sql);
                Long cost = System.currentTimeMillis() - start;
                log.info("插入数据{}条,耗时{}ms",update,cost);
            } catch (DataAccessException e) {
                /**
                 * 执行失败,设置回滚标志
                 *   这里用到了getAndSet() 会返回rollbackFlag被修改前的状态
                 *   因此在多个线程执行失败的时候
                 *   将只有一个线程会得到rollbackFlag为false
                 */
                boolean getPreStatus = rollbackFlag.getAndSet(true);
                if (!getPreStatus) {
                    rollbackFlag.set(true);
                    this.firstFail.add(e);
                    mainThreadLatch.countDown();
                }
            } finally {
                // 删除脏数据
                if (rollbackFlag.get()) {
                    jdbcTemplate.update("truncate table " + apiInfoDto.getTableName());
                }
                int i = successNumber.addAndGet(1);
                if (distributeNumber == i) {
                    mainThreadLatch.countDown();
                }
            }
            return true;
        }
    }

    /**
     * 任务返回类的包装对象
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Accessors(chain = true)
    class TaskReturn {
        //表名称
        private String tableName;
        //影响行数
        private Integer rows;
        //执行时间
        private Long time;
        //执行是否成功
        private Boolean isSuccess;
        //失败抛出的异常
        private Throwable exception;

    }
}







